import org.apache.spark.sql.SparkSession

import java.util.regex.Pattern

object data1_movies {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("Spark Pi")
      .getOrCreate()
    val sc = spark.sparkContext
    //电影数据分析
    val filePath = "F:/Spark/karry/karry/src/main/resources/"
    val moviesRDD = sc.textFile(filePath + "movies.dat")
    val occupationsRDD = sc.textFile(filePath + "occupations.dat")
    val ratingsRDD = sc.textFile(filePath + "ratings.dat")
    val usersRDD = sc.textFile(filePath + "users.dat")
    println("电影总数:" + moviesRDD.count() + "\n" + "用户总数:" + usersRDD.count())
    println("评分总数:" + ratingsRDD.count() + "\n" + "职业数:" + occupationsRDD.count())
    //统计某用户评分的电影总数及对应的电影详情（Id,电影名,类型）
//    val num = ratingsRDD.map(_.split("::"))
//      .map(user => (user(1), user(0))) //转换成key value型 电影ID 用户ID
//      .filter(_._2.equals("3888"))
//    println("用户Id为3888的评分电影总数为:" + num.count()) //用户id为3888评分电影总数
//    //所评分电影对于的详情(电影名，电影类型)
//    val moviesInfo = moviesRDD.map(_.split("::"))
//      .map(movies => (movies(0), (movies(1), (movies, (2)))))
//    val res = num.join(moviesInfo)
//      .map(item => (item._1, item._2._1, item._2._2._1, item._2._2._2))
//    println("电影详情是: \n")
//    res.take(3).foreach(println)




    //所有电影中平均得分最高的前 10 部电影
    // "ratings.dat"：UserID::MovieID::Rating::Timestamp
    // [UserID,MovieID,Rating,Timestamp]

    val rating = ratingsRDD.map(x => x.split("::")).map {
      x => {
        (x(0), x(1), x(2)) // (UserID,MovieID,Rating)
      }
    }.cache()
    println("平均得分最高的前 10 名的电影名称简单版")
    //  ( MovieId,( Rating, 1) ) (1200,(4.0,1))
    rating.map(x => (x._2, (x._3.toDouble, 1)))
      //  ( MovieId,( 总分, 总次数) )
      .reduceByKey((x, y) => {
        (x._1 + y._1, x._2 + y._2)
      })
      // (  平均分, MovieId )
      .map(x => (x._2._1 / x._2._2, x._1))
      .sortByKey(false)
      .take(5)
      .foreach(println)
    println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次 数))")
    // MovieID::Title::Genres
    val moviesInfo = moviesRDD.map(x => x.split("::"))
     // .filter(arr=>arr.length >=3)  //new
      .map(x => {
        (x(0), (x(1), x(2)))
      })
    val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
      // ( MovieId,( Rating, 1) )(1252,(4.0,1))
      .reduceByKey((x, y) => {
        (x._1 + y._1, x._2 + y._2)
      })
      .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2)))
    moviesInfo.join(ratingsInfo)
      .map(info => {
        (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
      }).sortByKey(false) // (平均分,(movieId,Title,Genres,总分,总次数))
      .take(5)
      .foreach(println)
    println("观影人数最多的前 10 部电影")
    // (UserID,MovieID,Rating)
    val watchViewsInfo = rating.map(x => {
        (x._2, 1)
      }).reduceByKey((x, y) => x + y)
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(5)
    // 5 名
    watchViewsInfo.foreach(println(_))
    println("===================>")

    rating.map(x => (x._2, 1))
      .reduceByKey((x, y) => {
        (x + y)
      }) //( MovieId,总次数 )
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(10)
      .foreach(println) //  286-> 999
    println("详情的输出(  观影人数，电影编号)")
    // 输出 ( 总次数, (MovieID, title,Genres,总分,平均分  ) )
    //  ratingsInfo ( MovieId, (平均分, 总分,总次数) )
    //  ( MovieID,  ((Title,Genres),(平均分, 总分,总次数)  ) )
    // ratingsInfo ( MovieId, (平均分, 总分,总次数) )
    // moviesInfo  MovieID::Title::Genres
    moviesInfo.join(ratingsInfo).map(x => {
        (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
      }).sortByKey(false)
      .take(10)
      .foreach(println)

    //统计不同电影类型的数量
    val movieStyle =moviesRDD.map(_.split("::"))
      .map(tp=>(tp(0),tp(2)))
      .flatMapValues(sp=>sp.split("\\|"))
      .map(p=>(p._2,1))
      .reduceByKey((x,y) =>x+y)
      .filter(_._1.equals("Comedy"))//打印喜剧片的数量
    movieStyle.foreach(println)
      //movieStyle.take(5).foreach(println)

    //正则表达式
    val pattern=Pattern.compile("(.*)(\\(\\d{4}\\))")
    val movieInfo = moviesRDD.map(_.split("::"))
      .map(x => (x(1), 1))
      .map(x => {
        var name = ""
        var year = ""
        val matcher = pattern.matcher(x._1)
        if (matcher.find()) {
          name = matcher.group(1)
          year = matcher.group(2)
          year = year.substring(1, year.length - 1)
        }
        if (year == "") {
          (-1, 1)
        } else {
          (year.toInt, 1)
        }
      })
      .reduceByKey((x,y) =>x+y)
      .sortByKey()
      .collect()
    movieInfo.take(10).foreach(println(_))


    val count2000=moviesRDD
      .map(_.split("::")(1))
      .filter(title =>title.contains("(2000)")).count()
    println(s"2000电影数量$count2000")
    // 从电影数据中提取年份并统计2000年的数量
//    val year2000Count = moviesRDD.map { line =>
//      val fields = line.split("::")
//      val title = fields(1) // 假设数据格式为 MovieID::Title::Genres
//      title
//    }.flatMap { title =>
//      // 使用正则表达式从标题中提取年份（如 "Title (2000)"）
//      val yearPattern = """\(\d{4}\)""".r
//      yearPattern.findFirstIn(title) // 查找括号内的4位数字（如 "(2000)"）
//        .map(_.replaceAll("[()]", "")) // 去掉括号，保留纯数字
//        .filter(_.toInt == 2000) // 过滤出2000年的记录
//    }.count() // 统计总数
//
//    println(s"\n2000年生产的电影数量: $year2000Count")

    sc.stop()

  }

}
